package com.study.spring.hadoop.hdfs;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.*;
import org.apache.hadoop.io.IOUtils;
import org.springframework.beans.factory.annotation.Autowired;

import java.io.FileOutputStream;
import java.io.InputStream;
import java.util.ArrayList;
import java.util.List;

public class FileSystemDemo {

    @Autowired
    Configuration configuration;

    /**
     * （1）获取hdfs当前路径下不是目录的文件
     *
     * @param path
     * @return
     */
    public List<HdfsFile> getFiles(String path) {
        try {
            FileSystem fs = FileSystem.get(configuration);
            RemoteIterator<LocatedFileStatus> fileStatusRemoteIterator = fs.listFiles(new Path(path), false);

            List<HdfsFile> files = new ArrayList<>();

            while (fileStatusRemoteIterator.hasNext()) {
                LocatedFileStatus fileStatus = fileStatusRemoteIterator.next();
                Path path1 = fileStatus.getPath();
                long blockSize = fileStatus.getBlockSize();
                String name = path1.getName();
                String uriPath = path1.toUri().getPath();
                String owner = fileStatus.getOwner();

                HdfsFile file = new HdfsFile();
                file.setFilePath(uriPath);
                file.setBlockSize(blockSize);
                file.setFileName(name);
                file.setFileOwner(owner);

                files.add(file);
            }
            return files;
        } catch (Exception e) {
            e.printStackTrace();
            return null;
        }
    }


    /**
     * （2）查询当前路径下所有的目录
     *
     * @param path
     * @return
     */
    public List<HdfsFile> getDirectory(String path) {
        try {
            FileSystem fs = FileSystem.get(configuration);
            FileStatus[] fileStatuses = fs.listStatus(new Path(path));

            List<HdfsFile> files = new ArrayList<>();
            for (FileStatus fileStatus : fileStatuses) {
                if (fileStatus.isDirectory()) {

                    Path path1 = fileStatus.getPath();

                    String name = path1.getName();
                    String uriPath = path1.toUri().getPath();
                    long blockSize = fileStatus.getBlockSize();
                    String owner = fileStatus.getOwner();

                    HdfsFile file = new HdfsFile();
                    file.setFilePath(uriPath);
                    file.setBlockSize(blockSize);
                    file.setFileName(name);
                    file.setFileOwner(owner);
                    files.add(file);
                }
            }

            return files;
        } catch (Exception e) {
            e.printStackTrace();
            return null;
        }
    }


    /**
     * （4）上传文件
     *
     * @param srcPath
     * @param destPath
     * @return
     */
    public Boolean uploadFile(String srcPath, String destPath) {
        try {
            FileSystem fs = FileSystem.get(configuration);
            fs.copyFromLocalFile(new Path(srcPath), new Path(destPath));
        } catch (Exception e) {
            e.printStackTrace();
            return false;
        }
        return true;
    }


    /**
     * （5）使用流上传文件
     *
     * @param inStream
     * @param hdfsPath
     * @return
     */
    public Boolean uploadFile2(InputStream inStream, String hdfsPath) {
        FSDataOutputStream fos = null;
        try {
            FileSystem fs = FileSystem.get(configuration);
            fos = fs.create(new Path(hdfsPath));
            IOUtils.copyBytes(inStream, fos, 4078);
        } catch (Exception e) {
            e.printStackTrace();
            return false;
        } finally {
            IOUtils.closeStream(fos);
        }
        return true;
    }


    /**
     * （6）使用流下载文件
     *
     * @param localPath
     * @param hdfsPath
     * @return
     */
    public Boolean downLoadFile(String localPath, String hdfsPath) {
        FSDataInputStream fis = null;
        FileOutputStream fos = null;
        try {
            FileSystem fs = FileSystem.get(configuration);
            fis = fs.open(new Path(hdfsPath));
            fos = new FileOutputStream(localPath);
            IOUtils.copyBytes(fis, fos, 4078);
        } catch (Exception e) {
            e.printStackTrace();
            return false;
        } finally {
            IOUtils.closeStream(fos);
            IOUtils.closeStream(fis);
        }
        return true;
    }


    /**
     * （7）写文件
     *
     * @param inpath
     * @param outpath
     * @return
     */
    public Boolean writeFile(String inpath, String outpath) {
        FSDataOutputStream fos = null;
        FSDataInputStream fis = null;
        try {
            FileSystem fs = FileSystem.get(configuration);
            fos = fs.create(new Path(outpath));
            fis = fs.open(new Path(inpath));
            IOUtils.copyBytes(fis, fos, 1024);
        } catch (Exception e) {
            e.printStackTrace();
            return false;
        } finally {
            IOUtils.closeStream(fis);
            IOUtils.closeStream(fos);
        }
        return true;
    }


    /**
     * （8）修改文件名
     *
     * @param oldPath
     * @param newPath
     * @return
     */
    public boolean renameFile(String oldPath, String newPath) {
        try {
            FileSystem fs = FileSystem.get(configuration);
            boolean flag = fs.rename(new Path(oldPath), new Path(newPath));
            return flag;
        } catch (Exception e) {
            e.printStackTrace();
            return false;
        }
    }

    /**
     * （9）移动hdfs上面的文件
     *
     * @param oldPath
     * @param newPath
     * @return
     */
    public boolean moveFile(String oldPath, String newPath) {
        FSDataInputStream fdis = null;
        FSDataOutputStream fos = null;
        try {
            FileSystem fs = FileSystem.get(configuration);
            fdis = fs.open(new Path(oldPath));
            fos = fs.create(new Path(newPath));

            IOUtils.copyBytes(fdis, fos, 4096);
            fs.delete(new Path(oldPath), true);
        } catch (Exception e) {
            e.printStackTrace();
            return false;
        } finally {
            IOUtils.closeStream(fos);
            IOUtils.closeStream(fdis);
        }
        return true;
    }

    /**
     * （10）hdfs移动目录，请小伙伴们注意咯：FileStatus只能读到当前一级文件或者目录，所以需要递归遍历才能读到所有的子文件
     *
     * @param oldPath
     * @param newPath
     * @return
     */
    public boolean moveDir(String oldPath, String newPath) {
        FSDataInputStream fdis = null;
        FSDataOutputStream fos = null;
        try {
            FileSystem fs = FileSystem.get(configuration);
            Path pathdir = new Path(oldPath);
            FileStatus[] fileStatuses = fs.listStatus(pathdir);

            String basePath = newPath + "/";
            //创建基础目录
            fs.mkdirs(new Path(basePath));

            for (FileStatus fileStatus : fileStatuses) {
                recuerFile(fdis, fos, fs, fileStatus, basePath);
            }
            fs.delete(new Path(oldPath), true);
        } catch (Exception e) {
            e.printStackTrace();
            return false;
        } finally {
            IOUtils.closeStream(fos);
            IOUtils.closeStream(fdis);
        }
        return true;
    }

    /**
     * @param fdis
     * @param fos
     * @param fs
     * @param fileStatus
     * @param basePath
     */
    public void recuerFile(FSDataInputStream fdis, FSDataOutputStream fos, FileSystem fs, FileStatus fileStatus, String basePath) {
        try {
            boolean dir = fileStatus.isDirectory();

            FileStatus[] fileStatuses = fs.listStatus(fileStatus.getPath());
            Path path = fileStatus.getPath();
            String fileName = path.getName();

            if (dir) {
                fs.mkdirs(new Path(basePath + fileName));

                //因为当前遍历到的是一个目录 ， basePath又要改成当前目录的路径，之后的文件写入这个新的目录
                basePath = basePath + "/" + fileName + "/";

                for (FileStatus fileStatus1 : fileStatuses) {
                    recuerFile(fdis, fos, fs, fileStatus1, basePath);
                }
            } else {
                fdis = fs.open(path);
                fos = fs.create(new Path(basePath + fileName));
                IOUtils.copyBytes(fdis, fos, 4096);
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }


    /**
     * （11）hdfs移动目录 ，使用callback减少业务代码入侵我们的方法！！
     * 1.首先创建一个回调接口Callback。里面就只有一个方法 void call(T obj);
     * 2.移动目录的逻辑
     */
    public boolean moveDir0(String oldPath, String newPath) {
        try {
            FileSystem fs = FileSystem.get(configuration);
            Path pathdir = new Path(oldPath);
            FileStatus[] fileStatuses = fs.listStatus(pathdir);

            final String basePath = newPath + "/";

            //创建基础目录
            fs.mkdirs(new Path(basePath));
            for (FileStatus fileStatus : fileStatuses) {
                //递归调用回调之后的业务
                recuerFile0(fs, fileStatus, new Callback<FileStatus>() {

                    String common_path = basePath;

                    @Override
                    public void call(FileStatus fileStatus) {
                        boolean dir = fileStatus.isDirectory();
                        Path path = fileStatus.getPath();
                        String fileName = path.getName();
                        FSDataInputStream fis = null;
                        FSDataOutputStream fos = null;

                        try {
                            if (dir) {
                                fs.mkdirs(new Path(common_path + path.toUri().getPath()));
                            } else {
                                fis = fs.open(path);
                                fos = fs.create(new Path(common_path + path.toUri().getPath()));
                                IOUtils.copyBytes(fis, fos, 4096);
                            }
                        } catch (Exception e) {
                            e.printStackTrace();
                        } finally {
                            IOUtils.closeStream(fos);
                            IOUtils.closeStream(fis);
                        }
                    }
                });
            }
            fs.delete(new Path(oldPath), true);
        } catch (Exception e) {
            e.printStackTrace();
            return false;
        }
        return true;
    }


    /**
     * 3.递归方法改版，删掉业务代码，就很清楚了^_^
     *
     * @param fs
     * @param fileStatus
     * @param callback
     */
    public void recuerFile0(FileSystem fs, FileStatus fileStatus, Callback callback) {
        try {
            boolean dir = fileStatus.isDirectory();
            callback.call(fileStatus);
            if (dir) {
                FileStatus[] fileStatuses = fs.listStatus(fileStatus.getPath());
                for (FileStatus fileStatus1 : fileStatuses) {
                    recuerFile0(fs, fileStatus1, callback);
                }
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }


    /**
     * （12）hdfs删除文件，如果是用hdfs的API的话，默认删除的文件不会放入到回收站，需要自己去实现这个功能！！！
     *
     * @param path
     * @return
     */
    public boolean deleteFile(String path) {
        try {
            FileSystem fs = FileSystem.get(configuration);
            boolean flag = fs.delete(new Path(path), true);
            return flag;
        } catch (Exception e) {
            e.printStackTrace();
            return false;
        }
    }


    /**
     * （13）删除文件，并且将文件放入到回收站
     *
     * @param path
     * @return
     */
    public boolean deleteFile2Trash(String path) {
        try {
            Trash trash = new Trash(configuration);
            boolean tflag = trash.moveToTrash(new Path(path));
            return tflag;
        } catch (Exception e) {
            e.printStackTrace();
            return false;
        }
    }

}
